package org.hope.lee.consumer.broadercast;

import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

public class ConsumerBroadCastMember2 {
	public static void main(String[] args) throws MQClientException {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_broadcast");
		consumer.setNamesrvAddr("192.168.31.176:9876;192.168.31.165:9876");
		// 批量消费,每次拉取10条
		consumer.setConsumeMessageBatchMaxSize(10);
		//设置广播消费
		consumer.setMessageModel(MessageModel.BROADCASTING);
		//设置集群消费
//		consumer.setMessageModel(MessageModel.CLUSTERING);
		// 如果非第一次启动，那么按照上次消费的位置继续消费
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		// 订阅PushTopic下Tag为push的消息
		consumer.subscribe("topic_broadcast", "TagA || Tag B || Tage C");
		consumer.registerMessageListener(new MqBroadCastListener());
		consumer.start();
		System.out.println("Consumer2 Started.");

	}
}
